package com.atguigu.flink.state.opearatestate;

import com.atguigu.flink.func.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by 黄凯 on 2023/6/20 0020 19:16
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 使用场景很局限： 用于两个流。
 * *                      一个是消费数据的，称为数据流。
 * *                      一个是定义规则，称为配置流。
 * *
 * *                  希望配置流中的规则更新时，可以把新的配置自动广播到数据流的每一个Task，可以使用广播状态。
 * *
 * *    ------------------
 * *      将新的规则放入广播状态对象即可。自动广播。
 */
public class Flink04_BroadCastState {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(2000);

        //准备两个流
        SingleOutputStreamOperator<WaterSensor> dataDS =
                env.socketTextStream("127.0.0.1", 8888)
                        .map(new WaterSensorMapFunction());

        SingleOutputStreamOperator<MyConf> configDS = env
                .socketTextStream("127.0.0.1", 9999)
                .map(line -> {
                    String[] words = line.split(",");
                    return new MyConf(words[0], words[1]);
                });

        //2.普通流无法使用广播状态，必须是广播流
        MapStateDescriptor<String, MyConf> mapStateDescriptor = new MapStateDescriptor<>("config", String.class, MyConf.class);

        BroadcastStream<MyConf> broadcastStream = configDS.broadcast(mapStateDescriptor);

        //3.让两个流产生联系
        BroadcastConnectedStream<WaterSensor, MyConf> connectedStream = dataDS.connect(broadcastStream);

        connectedStream
                .process(
                        new BroadcastProcessFunction<WaterSensor, MyConf, WaterSensor>() {

                            //处理数据流中的数据
                            @Override
                            public void processElement(WaterSensor waterSensor,
                                                       BroadcastProcessFunction<WaterSensor, MyConf, WaterSensor>.ReadOnlyContext context,
                                                       Collector<WaterSensor> out) throws Exception {

                                //获取广播状态，取出最新的配置
                                ReadOnlyBroadcastState<String, MyConf> broadcastState = context.getBroadcastState(mapStateDescriptor);
                                MyConf myConf = broadcastState.get(waterSensor.getId());

                                //使用配置
                                waterSensor.setId(myConf.getName());
                                out.collect(waterSensor);


                            }

                            /**
                             * 处理配置更新
                             */
                            @Override
                            public void processBroadcastElement(MyConf myConf,
                                                                BroadcastProcessFunction<WaterSensor, MyConf, WaterSensor>.Context context,
                                                                Collector<WaterSensor> collector) throws Exception {

                                //把新的配置，放入广播状态即可
                                BroadcastState<String, MyConf> broadcastState = context.getBroadcastState(mapStateDescriptor);

                                //把广播状态当map用
                                broadcastState.put(myConf.getId(), myConf);

                                //自动广播到数据的每一个task

                            }
                        }
                ).print();

        env.execute();

    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyConf {
        private String id;
        private String name;
    }

}
